package com.baitiaojun.bulletscreen.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.baitiaojun.api.domain.video.BulletScreen;
import com.baitiaojun.api.domain.video.BulletScreenMqDuplicate;
import com.baitiaojun.api.exception.ConditionException;
import com.baitiaojun.api.util.SnowFlakeUtil;
import com.baitiaojun.bulletscreen.config.properties.MqBulletScreenProperties;
import com.baitiaojun.bulletscreen.config.properties.MqSyncMysqlProperties;
import com.baitiaojun.bulletscreen.websocket.BulletScreenWebSocketServer;
import com.baitiaojun.bulletscreen.service.BulletScreenMqDuplicateService;
import com.baitiaojun.bulletscreen.service.BulletScreenService;
import com.baitiaojun.bulletscreen.service.MqBulletScreenService;
import com.baitiaojun.bulletscreen.support.RedisSupport;
import com.baitiaojun.bulletscreen.util.BulletScreenConstantUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @类名 MqBulletScreenServiceImpl
 * @描述 TODO
 * @作者 白条君
 * @创建日期 2022/12/31 18:11
 * @版本 1.0
 */
@Transactional
@Slf4j
@Service("mqBulletScreenServiceImpl")
public class MqBulletScreenServiceImpl implements MqBulletScreenService {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private MqBulletScreenProperties mqBulletScreenProperties;

    @Resource
    private MqSyncMysqlProperties mqSyncMysqlProperties;

    @Autowired
    private BulletScreenService bulletScreenService;

    @Autowired
    private RedisSupport redisSupport;

    @Resource
    private BulletScreenMqDuplicateService duplicateService;
    
    private static final AtomicInteger COUNT = new AtomicInteger(0);

    @Retryable(maxAttempts = 3, backoff = @Backoff(delay = 5 * 1000, multiplier = 3))
    @Async
    @Override
    public void bulletScreenSender(BulletScreen bulletScreen, String sessionId) {
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(String.valueOf(bulletScreen.getId()));
        Message message = MessageBuilder
                .withBody(JSONObject.toJSONString(bulletScreen)
                .getBytes(StandardCharsets.UTF_8))
                .build();
        MessageProperties messageProperties = message.getMessageProperties();
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        messageProperties.setMessageId(sessionId);
        rabbitTemplate.convertAndSend(mqBulletScreenProperties.getExchange()
                , mqBulletScreenProperties.getRoutingKey()
                , message
                , correlationData
        );
    }

    @Async
    @Override
    @RabbitListener(queues = {"mirror_bullet_screen_queue", "mirror_backup_bullet_screen_queue"}, containerFactory = "listenerContainerFactory")
    public void bulletScreenReceiver(Message message, Channel channel) throws IOException {
        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();
        //获取webSocketHandler实例
        Map<String, BulletScreenWebSocketServer> bulletScreenMap = BulletScreenConstantUtil.BULLET_SCREEN_MAP;
        //获取bulletScreen的id和content
        BulletScreen bulletScreen = JSONObject.parseObject(new String(message.getBody()), BulletScreen.class);
        String key = "send-bullet-screen" + bulletScreen.getId();
        String msg = bulletScreen.getContent();
        try {
            //发送弹幕到前端
            if (redisSupport.setNx(key, "", 1L, TimeUnit.MINUTES)) {
                BulletScreenWebSocketServer webSocketHandler = bulletScreenMap.get(messageProperties.getMessageId());
                if (webSocketHandler.getSession().isOpen()) {
                    webSocketHandler.sendMsg(msg);
                    channel.basicAck(deliveryTag, false);
                    COUNT.getAndIncrement();
                    log.info("收到消息：{}，总共消费{}条消息", msg, COUNT.get());
                }
            }
        } catch (Exception e) {
            channel.basicNack(deliveryTag, false, false);
            log.error(e.getMessage());
        }
    }

    @Async
    @Override
    @RabbitListener(queues = {"mirror_dead_bullet_screen_queue"}, containerFactory = "listenerContainerFactory")
    public void deadBulletScreenReceiver(Message message, Channel channel) {
        MessageProperties messageProperties = message.getMessageProperties();
        //获取webSocketHandler实例
        Map<String, BulletScreenWebSocketServer> bulletScreenMap = BulletScreenConstantUtil.BULLET_SCREEN_MAP;
        //获取bulletScreen的id和content
        BulletScreen bulletScreen = JSONObject.parseObject(new String(message.getBody()), BulletScreen.class);
        String key = "send-bullet-screen" + bulletScreen.getId();
        String msg = bulletScreen.getContent();
        try {
            //发送弹幕到前端
            if (redisSupport.setNx(key, "", 1L, TimeUnit.MINUTES)) {
                BulletScreenWebSocketServer webSocketHandler = bulletScreenMap.get(messageProperties.getMessageId());
                if (ObjectUtils.isNotEmpty(webSocketHandler) && webSocketHandler.getSession().isOpen()) {
                    webSocketHandler.sendMsg(msg);
                    channel.basicAck(messageProperties.getDeliveryTag(), false);
                    COUNT.getAndIncrement();
                    log.info("收到消息：{}", bulletScreen.getContent());
                }
            }
        } catch (Exception e) {
            log.error("死信队列消费失败");
            //短信发送提示是哪条消息出现问题，进行人工处理

        }
    }

    @Override
    @RabbitListener(queues = "mirror_warning_bullet_screen_queue")
    public void bulletScreenReceiverWarning(Message message) {
        log.warn("警告！路由不可达，交由备份交换机和备份队列处理，消息：{}"
                , JSONObject.parseObject(new String(message.getBody()), BulletScreen.class).getContent());
    }

    @Retryable(maxAttempts = 3, backoff = @Backoff(delay = 5 * 1000, multiplier = 3))
    @Async
    @Override
    public void syncMysqlSender(BulletScreen bulletScreen) throws ConditionException {
        CorrelationData correlationData = new CorrelationData();
        Long id = SnowFlakeUtil.getNextId();
        correlationData.setId(String.valueOf(id));
        String msg = JSONObject.toJSONString(bulletScreen);
        Message message = MessageBuilder.withBody(msg.getBytes(StandardCharsets.UTF_8)).build();
        MessageProperties messageProperties = message.getMessageProperties();
        messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        //冗余存储
        duplicateService.saveBulletScreenMqDuplicate(
                BulletScreenMqDuplicate.builder()
                        .id(id)
                        .content(msg)
                        .status("0")
                        .createtime(new Date())
                        .build()
        );
        rabbitTemplate.convertAndSend(mqSyncMysqlProperties.getSyncMysqlExchange(), mqSyncMysqlProperties.getSyncMysqlRoutingKey(), message, correlationData);
    }

    @Async
    @Override
    @RabbitListener(queues = {"mirror_sync_mysql_queue", "mirror_backup_sync_mysql_queue"}, containerFactory = "listenerContainerFactory")
    public void syncMysqlReceiver(Message message, Channel channel) throws IOException, ConditionException {
        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();
        BulletScreen bulletScreen = JSONObject.parseObject(new String(message.getBody()), BulletScreen.class);
        String key = "sync-mysql" + bulletScreen.getId();
        if (redisSupport.setNx(key, "", 1L, TimeUnit.MINUTES)) {
            //存储到mysql
            bulletScreenService.saveBulletScreen(bulletScreen);
        }
        try {
            channel.basicAck(deliveryTag, false);
            log.info("收到消息：{}", bulletScreen.getContent());
        } catch (Exception e) {
            channel.basicNack(deliveryTag, false, false);
            log.error(e.getMessage());
        }
    }

    @Async
    @Override
    @RabbitListener(queues = {"mirror_dead_sync_mysql_queue"}, containerFactory = "listenerContainerFactory")
    public void deadSyncMysqlReceiver(Message message, Channel channel) throws ConditionException {
        MessageProperties messageProperties = message.getMessageProperties();
        BulletScreen bulletScreen = JSONObject.parseObject(new String(message.getBody()), BulletScreen.class);
        String key = "sync-mysql" + bulletScreen.getId();
        if (redisSupport.setNx(key, "", 1L, TimeUnit.MINUTES)) {
            //存储到mysql
            bulletScreenService.saveBulletScreen(bulletScreen);
        }
        try {
            channel.basicAck(messageProperties.getDeliveryTag(), false);
            log.info("收到消息：{}", bulletScreen.getContent());
        } catch (Exception e) {
            log.error("死信队列消费失败");
            //短信发送提示是哪条消息出现问题，进行人工处理

        }
    }

    @Override
    @RabbitListener(queues = "mirror_warning_sync_mysql_queue")
    public void syncMysqlReceiverWarning(Message message) {
        log.warn("警告！路由不可达，交由备份交换机和备份队列处理，消息：{}",
                JSONObject.parseObject(new String(message.getBody()), BulletScreen.class).getContent());
    }
}
